package com.example.mq.mqserver.core;

import com.example.mq.common.Consumer;
import com.example.mq.common.ConsumerEnv;
import com.example.mq.common.MqException;
import com.example.mq.mqserver.VirtualHost;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;

// 实现消费者核心逻辑
public class ConsumerManager {
    // 持有上层VirtualHost对象的引用，用来操作数据
    private VirtualHost parent;
    // 指定一个线程池，执行具体的任务
    private ExecutorService workerPool = Executors.newFixedThreadPool(4);
    // 存放令牌的队列
    private BlockingQueue<String> tokenQueue = new LinkedBlockingQueue<>();
    // 扫描线程
    private Thread scanThread = null;
    public ConsumerManager(VirtualHost parent) {
        this.parent = parent;

        scanThread = new Thread(() -> {
            while (true) {
                try {
                    // 拿到令牌
                    String queueName = tokenQueue.take();
                    // 找到对应的队列
                    MSGQueue queue = parent.getMemoryDataCenter().getQueue(queueName);
                    if(queue == null) {
                        throw new MqException("[consumerManager] 取令牌后，队列不存在！queueName=" + queueName);
                    }
                    // 执行消费
                    synchronized (queue) {
                        consumeMessage(queue);
                    }
                } catch (InterruptedException | MqException e) {
                    throw new RuntimeException(e);
                }
            }
        });
        // 把线程设为后台线程
        scanThread.setDaemon(true);
        scanThread.start();
    }

    // 通知消费，调用时机是在发送消息的时候
    public void notifyConsumer(String queueName) throws InterruptedException {
        tokenQueue.put(queueName);
    }

    public void addConsumer(String consumerTag, String queueName, boolean autoAck, Consumer consumer) throws MqException {
        // 找到对应的队列
        MSGQueue queue = parent.getMemoryDataCenter().getQueue(queueName);
        if(queue == null) {
            throw new MqException("队列不存在! queueName:" + queueName);
        }
        ConsumerEnv consumerEnv = new ConsumerEnv(consumerTag,queueName,autoAck,consumer);
        synchronized (queue){
            queue.addConsumerEnv(consumerEnv);
            // 如果当前队列已经有了一些消息，就要立即消费掉
            int n = parent.getMemoryDataCenter().getMessageCount(queueName);
            for (int i = 0; i < n; i++) {
                // 调用一次，消费一条消息
                consumeMessage(queue);
            }
        }
    }

    private void consumeMessage(MSGQueue queue) {
        // 1.轮询找消费者
        ConsumerEnv luckyDog = queue.chooseConsumer();
        if(luckyDog == null) {
            // 没有消费者，暂时不消费
            return;
        }
        // 2.从队列中取出一条消息
        Message message = parent.getMemoryDataCenter().pollMessage(queue.getName());
        if(message == null) {
            // 没有消息，暂时不消费
            return;
        }
        // 3.把消息带入消费者的回调方法，丢给线程池执行
        workerPool.submit(() -> {
            try {
                // 1.把消息放到待确认的集合，这个操作在回调之前
                parent.getMemoryDataCenter().addMessageWaitAck(queue.getName(), message);
                // 2.调用消费者的回调方法
                luckyDog.getConsumer().handleDelivery(luckyDog.getConsumerTag(),
                        message.getBasicProperties(), message.getBody());
                // 3.“自动应答”，可以直接删除
                //   “手动应答”，先不进行处理，交给后续消费者调用basicAck方法
                if(luckyDog.isAutoAck()) {
                    // 硬盘
                    if(message.getDeliveryMode() == 2) {
                        parent.getDiskDataCenter().deleteMessage(queue,message);
                    }
                    // 上面待确认集合中消息
                    parent.getMemoryDataCenter().removeMessageWaitAck(queue.getName(), message.getMessageId());
                    // 内存中消息中心的消息
                    parent.getMemoryDataCenter().removeMessage(message.getMessageId());
                    System.out.println("[ConsumerManger] 消息被成功消费！queueName=" + queue.getName());
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        });
    }
}
